package cn.doitedu.hbase.day01;

import org.apache.hadoop.hbase.TableName;
import org.apache.hadoop.hbase.client.Connection;
import org.apache.hadoop.hbase.client.Put;
import org.apache.hadoop.hbase.client.Table;
import org.apache.hadoop.hbase.util.Bytes;

import java.io.BufferedReader;
import java.io.File;
import java.io.FileReader;
import java.io.IOException;
import java.util.ArrayList;

/**
 * 将数据批量写入到Hbase中
 */
public class PutsDemo {

    public static void main(String[] args) {

        Connection connection = null;
        Table table = null;
        try {
            connection = ConnectionUtils.getConnection();
            table = connection.getTable(TableName.valueOf("tb_users2"));
            ArrayList<Put> puts = new ArrayList<>(1000);

            //读取指定文件中的数据
            BufferedReader bufferedReader = new BufferedReader(new FileReader("data/students.txt"));
            String line = null;

            while ((line = bufferedReader.readLine()) != null) {
                String[] fields = line.split(",");
                String id = fields[0];
                String name = fields[1];
                int age = Integer.parseInt(fields[2]);
                String gender = fields[3];
                String city = fields[4];
                double sal = Double.parseDouble(fields[5]);
                String dt = fields[6];

                Put put = new Put(Bytes.toBytes(id));
                put.addColumn(Bytes.toBytes("f1"), Bytes.toBytes("name"), Bytes.toBytes(name));
                put.addColumn(Bytes.toBytes("f1"), Bytes.toBytes("age"), Bytes.toBytes(age));
                put.addColumn(Bytes.toBytes("f1"), Bytes.toBytes("gender"), Bytes.toBytes(gender));
                put.addColumn(Bytes.toBytes("f2"), Bytes.toBytes("city"), Bytes.toBytes(city));
                put.addColumn(Bytes.toBytes("f2"), Bytes.toBytes("sal"), Bytes.toBytes(sal));
                put.addColumn(Bytes.toBytes("f1"), Bytes.toBytes("dt"), Bytes.toBytes(dt));
                //先将put对象攒起来（ArrayList）
                puts.add(put);
                if (puts.size() == 1000) {
                    table.put(puts); //匹配写入
                    puts.clear(); //清空数据
                }
            }
            //将条数不满足1000的数据再写入一次
            table.put(puts); //匹配写入

        } catch (Exception e) {
            e.printStackTrace();
        } finally {
            if (table != null)
                try {
                    table.close();
                } catch (IOException e) {
                    e.printStackTrace();
                }
            if (connection != null) {
                try {
                    connection.close();
                } catch (IOException e) {
                    e.printStackTrace();
                }
            }
        }


    }
}
